package net.bwie.flink;

import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.concurrent.TimeUnit;

/**
 * Flink DataStream采用异步方式，对流中数据进行补全，访问MySQL维表数据
 *      https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/datastream/operators/asyncio/
 * @author xuanyu
 * @date 2025/10/29
 */
public class _04OrderDimASync {

	public static void main(String[] args) throws Exception{
		// 1. 执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		/*
模拟数据：
u101,o1,milk,5.0,2025-10-29 17:30:10
u102,o2,water,2.0,2025-10-29 17:31:23
u103,o3,bread,6.0,2025-10-29 17:32:45
u104,o4,smoke,17.0,2025-10-29 17:33:17
        */
		DataStreamSource<String> stream = env.socketTextStream("node101", 9999);

		// 3. 数据转换-transformation
		DataStream<String> resultStream = AsyncDataStream.unorderedWait(
			stream,
			new AsyncDatabaseRequest(),
			1000,
			TimeUnit.MILLISECONDS,
			100
		);
		resultStream.print("async") ;

		// 5. 触发执行-execute
		env.execute("OrderDimASync") ;
	}

}

